/**
 * Alipay.com Inc.
 * Copyright (c) 2004-2022 All Rights Reserved.
 */
package com.example.spider.proxy.provider.local;

import com.example.spider.proxy.core.ProxyInfo;
import com.example.spider.proxy.core.tester.ProxyTypeEnum;
import com.example.spider.proxy.provider.AbstractProxyResultHandler;
import com.example.spider.proxy.provider.IpProxySpiderTask;
import com.example.spider.starter.SpiderStarter;
import com.example.spider.task.SpiderTaskQueue;
import com.example.spider.task.limit.FixWindowLimit;
import com.example.spider.web.ProgressController;
import lombok.Data;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.StringCodec;
import org.shoulder.batch.progress.BatchProgress;
import org.shoulder.batch.progress.BatchProgressCache;
import org.shoulder.core.util.JsonUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * redis -> repository
 *
 * @author lym
 */
//@Component
public class RedisProxyInfoHandler extends AbstractProxyResultHandler implements SpiderStarter {

    @Lazy
    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private BatchProgressCache progressCache;

    public RedisProxyInfoHandler() {
        super("local_redis");
    }

    protected Duration defaultLimitRateDuration() {
        return Duration.ofSeconds(5);
    }

    @Override
    public void startSpiders() {
        // 仅触发，不设置参数
        IpProxySpiderTask task = new IpProxySpiderTask()
                .setIpProxyTaskId(taskName)
                .setLimitStrategy(new FixWindowLimit(taskName, defaultLimitRateDuration()))
                .setIdempotentFunction(t -> "USE_REDIS")
                .setUrl("https://www.abc.com");
        SpiderTaskQueue.putTask(task);
    }

    /**
     * 每 10 分钟触发一次
     */
    @Override
    public String executeCron() {
        return null;
    }

    /**
     * 处理响应
     */
    @Override
    public void doHandle(IpProxySpiderTask task, String textFile) {
        // 其他代理池爬的
//        Threads.execute(this::loadFromPyProxySpider);

//        Threads.execute(() -> loadFromRedisRepository(ProxyRepository.SOURCE_REPO_REDIS_KEY_DEFAULT, super::addProxyToSourceRepository));
//        Threads.execute(() -> loadFromRedisRepository(ProxyRepository.USABLE_REPO_REDIS_KEY_DEFAULT, super::addProxyToUsableRepository));
//        Threads.execute(() -> loadFromRedisRepository(ProxyRepository.STABLE_REPO_REDIS_KEY_DEFAULT, super::addProxyToStableRepository));
    }

    private void loadFromPyProxySpider() {
        AtomicInteger counter = new AtomicInteger();
        RMap<String, String> map = redissonClient.getMap("use_proxy", StringCodec.INSTANCE);
        map.values().parallelStream()
                .map(v -> JsonUtils.parseObject(v, RemoteProxyInfo.class))
                .map(this::convertToProxyInfo)
                .peek(p -> counter.incrementAndGet())
                .forEach(super::addProxyToUsableRepository);
        log.info("read from redis [use_proxy], total={}.", counter.get());
    }

    private void loadFromRedisRepository(String redisKey, Consumer<ProxyInfo> storeRepository) {
        RMap<String, Object> map = redissonClient.getMap(redisKey);

        BatchProgress progress = new BatchProgress();
        progress.setId("redisToSourceRepo_" + redisKey);
        progress.setTotal(map.size());
        progress.start();
        progressCache.triggerFlushProgress(progress);
        log.warn("redis -> repo({}) 开始同步，进度： {}", redisKey, ProgressController.getProgressUrl(progress.getId()));
        map.values().parallelStream()
                // 兼容debug class convert
                .map(p -> {
                    ProxyInfo px = new ProxyInfo();
                    BeanUtils.copyProperties(p, px);
                    return px;
                })
                .peek(storeRepository)
                .forEach(p -> progress.addSuccess(1));
        progress.finish();
        log.info("read from redis [{}], total={} cost={}.", redisKey, progress.getTotal(), Duration.between(progress.getStartTime(), progress.getStopTime()));
    }

    private ProxyInfo convertToProxyInfo(RemoteProxyInfo remote) {
        String[] ipPort = remote.getProxy().split(":");
        return new ProxyInfo()
                .setIp(ipPort[0])
                .setPort(Integer.parseInt(ipPort[1]))
                .setSource(remote.getSource())
                .setSupportHttps(remote.getHttps())
                .setProxyTypeEnum(remote.getHttps() ? ProxyTypeEnum.HTTPS : ProxyTypeEnum.HTTP)
                .setLocation(remote.getRegion())
                .setAnonymity(remote.getAnonymous())
//                .setLastVerificationTime(remote.getLast_time())
                .setTotalCount(remote.getCheck_count())
                .setSuccessCount(remote.getCheck_count() - remote.getFail_count())
                .setFailCount(remote.getFail_count())
                //.setSuccessPercentage(remote.getAnonymous())
                //.setScore()
                ;
    }

    /**
     * 加载 txt文件，主要用于兼容外部
     * ip:port
     * ip:port
     */
    @Data
    public static class RemoteProxyInfo {
        private String proxy;
        private String source;
        private Boolean https;
        private String region;
        private String anonymous;
        private Integer check_count;
        private Integer fail_count;
        private String last_time;
        private Boolean last_status;
    }


    // key 122.227.236.123:7302
    // {"proxy": "122.227.236.123:7302", "https": false, "fail_count": 0, "region": "", "anonymous": "", "source": "freeProxy10", "check_count": 72, "last_status": true, "last_time": "2022-09-14 15:14:28"}
}